package com.qdu.mr.hbase.reduce;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import java.io.IOException;

// 复用com.qdu.mr.CountMapper（这是传统的Mapper类）
// 由此Mapper类提供K-Collection<V>集合
// 借助HBase封装的Reducer将数据存入HBase
// 在此Reducer中，需要向外（HBase）输出RowKey和Put，因为框架会实现put命令，需要至少这2个参数
// context.write(RowKey,PUT)可以在一个reduce方法中多次调用
// 因为values集合对应的RowKey是同一个，所以，有可能需要多次调用put的addColumns方法，实现一次写入多个列

// done 实现TableReducer，为HBase提供规定格式的数据
public class CountHBaseReducer
    extends TableReducer<Text,LongWritable,ImmutableBytesWritable>
{
    // done 实现reduce方法，接收来自Mapper按照Key聚合过的所有数据，转换为PUT

    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

        Long count=0l;
        for(LongWritable val : values){
            count+=val.get();
        }

        Put put = new Put(key.toString().getBytes());
        put.addColumn(
                "cf1".getBytes(),
                "count".getBytes(),
                Bytes.toBytes(count)
        );

        context.write(
                new ImmutableBytesWritable(key.toString().getBytes()),
                put
        );
    }
}
